package com.deodar.kettle.platform.monitor.service;


import com.deodar.kettle.platform.common.App;
import com.deodar.kettle.platform.common.util.KettleConstant;
import com.deodar.kettle.platform.common.util.RepositoryUtils;
import com.deodar.kettle.platform.database.domain.RJob;
import com.deodar.kettle.platform.database.service.IRDirectoryService;
import com.deodar.kettle.platform.database.service.IRJobService;
import com.deodar.kettle.platform.monitor.executor.JobExecutor;
import com.deodar.kettle.platform.monitor.util.KettlePoolUtil;
import com.deodar.kettle.platform.monitor.util.TaskManageUtil;
import com.deodar.kettle.platform.parameter.service.IRnParameterService;
import lombok.extern.slf4j.Slf4j;
import org.pentaho.di.base.AbstractMeta;
import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.core.logging.KettleLoggingEvent;
import org.pentaho.di.core.logging.LoggingBuffer;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobExecutionConfiguration;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.ObjectId;
import org.pentaho.di.repository.Repository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Map;

/**
 * @author: Administrator
 * @Date: 2019/12/8 15:58
 * @Description:
 * @version:1.0.0
 */
@Slf4j
@Service("jobService")
public class JobService {

    @Autowired
    @Qualifier("directoryService")
    IRDirectoryService directoryService;

    @Autowired
    @Qualifier("rJobService")
    IRJobService rJobService;

    @Autowired
    IRnParameterService bnParameterService;


    public void  delete(Integer jobId){
        Repository repository = App.getInstance().getRepository();
        try {
            repository.deleteJob(RepositoryUtils.getObjectId(jobId));
        } catch (KettleException e) {
            e.printStackTrace();
        }

    }


    public boolean start(Integer id){
        return startJob(id);
    }

    /**
     * 启动远程服务Job
     * @param id
     * @param logId
     * @return
     */
    public boolean startJob(Integer id,String logId){
        long start = System.currentTimeMillis();
        RJob job = rJobService.selectRJobById(id);
        try{
            Repository repository = App.getInstance().getRepository();
            ObjectId objectId = RepositoryUtils.getObjectId(id);
            JobMeta jobMeta = repository.loadJob(objectId,null);
            JobExecutor jobExecutor = new JobExecutor(decode(null,bnParameterService.selectMapByGroupId(job.getParGroupId())),jobMeta,logId);
            KettlePoolUtil.getInstance().run(jobExecutor);
        } catch (KettleException e) {
            e.printStackTrace();
        }
        log.info("Job执行时间：{}",System.currentTimeMillis() - start);

        return true;
    }

    /**
     * 启动Repository Job
     * @param id
     * @return
     */
    public boolean startJob(Integer id){
        long start = System.currentTimeMillis();
        try{
            Repository repository = App.getInstance().getRepository();
            ObjectId objectId = RepositoryUtils.getObjectId(id);
            JobMeta jobMeta = repository.loadJob(objectId,null);
            Job job1 = new Job(repository,jobMeta);
            job1.setDaemon(true);
            // 开始执行kjb
            job1.start();
            // 线程等待，直到kjb执行完成
            job1.waitUntilFinished();

            StringBuilder eventBuffer = new StringBuilder(10000);
            LoggingBuffer appender = KettleLogStore.getAppender();
            List<KettleLoggingEvent> loggingEvents = appender.getLogBufferFromTo(job1.getLogChannelId(), true, 0, appender.getLastBufferLineNr()+1);
            loggingEvents.stream().filter(event -> event.timeStamp >  job1.getLogDate().getTime()-1000).forEach(event -> {
                eventBuffer.append(appender.getLayout().format(event)).append(Const.CR);
            });
            final String logText = eventBuffer.toString();
            // 判断执行过程中是否有错误
            if (job1.getErrors() > 0) {
                throw new KettleException(logText);
            }
        } catch (KettleException e) {
            e.printStackTrace();
        }
        log.info("Job执行时间：{}",System.currentTimeMillis() - start);

        return true;
    }


    public JobExecutionConfiguration decode(AbstractMeta meta, Map<String,String> mapVariables){

        JobExecutionConfiguration jobExecutionConfiguration = new JobExecutionConfiguration();
        if(mapVariables != null) {
            jobExecutionConfiguration.setVariables(mapVariables);
        }
        jobExecutionConfiguration.setExecutingRemotely(true);
        jobExecutionConfiguration.setRepository(App.getInstance().getRepository());
        return jobExecutionConfiguration;

    }



    public boolean stop(String logId){
        String key = KettleConstant.KEY_PREFIX+logId;
        JobExecutor transExecutor = (JobExecutor) TaskManageUtil.getInstance().get(key);
        if(transExecutor != null){
            transExecutor.stop();
            transExecutor.setClickStop(true);
            log.info("接收停止信号，开始停止job任务");
            return true;
        }
        return false;
    }


}
